package com.aispeech.task.api.service.impl;

import com.aispeech.constant.Progress;
import com.aispeech.constant.RedisKeys;
import com.aispeech.data.pojo.task.Metrics;
import com.aispeech.data.pojo.task.Request;
import com.aispeech.data.pojo.task.Task;
import com.aispeech.task.api.data.enums.Errors;
import com.aispeech.task.api.data.form.TaskForm;
import com.aispeech.task.api.data.vo.TaskProgress;
import com.aispeech.task.api.exceptions.StaticError;
import com.aispeech.task.api.service.ITaskService;
import com.aispeech.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.data.redis.core.ReactiveRedisOperations;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;

@Slf4j
@Service
public class TaskServiceImpl implements ITaskService {
    @Autowired
    private ReactiveRedisOperations<String, Task> taskOp;

    @Autowired
    private ReactiveRedisOperations<String, String> strOp;

    @Value("${spring.redis.prefix.task}")
    private String taskKeyPrefix;

    @Value("${spring.redis.prefix.product}")
    private String productKeyPrefix;

    private RedisScript luaProductInfo;

    @PostConstruct
    public void init() {
        Resource resAppVerifyLua = new ClassPathResource("lua/app_verify.lua");
        luaProductInfo = RedisScript.of(resAppVerifyLua, List.class);
    }

    @Override
    public Mono<TaskProgress> newTask(Request request) {
        String taskId = StringUtils.uuid();

        Metrics metrics = Metrics.builder().receive_t(System.currentTimeMillis()).build();

        Task task = Task.builder()
                .id(taskId)
                .request(request)
                .metrics(metrics)
                .build();

        String productId = task.getRequest().getProductId();

        return getProductSettingAndStats(task, productId).then(
                taskOp
                        .opsForHash()
                        .put(RedisKeys.HASH_TASK(taskKeyPrefix, productId), taskId, task)
                        .then(strOp.opsForZSet().add(RedisKeys.SORTEDSET_TASKs, String.format("%s.%s", productId, taskId), task.getPriority()))
                        .then(strOp.opsForHash().put(RedisKeys.HASH_TASK_PROGRESS, taskId, String.valueOf(task.getProgress())))
                        .map(
                                saved -> {
                                    if (saved) {
                                        return TaskProgress.builder().taskId(taskId).build();
                                    } else {
                                        log.error("save task: {} failed.", taskId);
                                        return null;
                                    }
                                }));
    }

    @Override
    public Mono<TaskProgress> progress(TaskForm taskForm) {
        String taskId = taskForm.getTaskId();

        return strOp.opsForHash()
                .get(RedisKeys.HASH_TASK_PROGRESS, taskForm.getTaskId())
                .map(value -> TaskProgress.builder().taskId(taskId).progress(Integer.valueOf((String)value)).build());
    }

    @Override
    public Mono<Task> result(TaskForm taskForm) {
        return taskOp.opsForHash()
                .get(RedisKeys.HASH_TASK(taskKeyPrefix, taskForm.getProductId()), taskForm.getTaskId())
                .map(o -> {
                    Task task = (Task) o;

                    if (task.getProgress() < Progress.FINISHED) {
                        throw new StaticError(Errors.ERR_TASK_NOT_FINISHED);
                    } else {
                        return task;
                    }
                });
    }

    private Mono<Void> getProductSettingAndStats(Task task, String productId) {
        return strOp
                .execute(luaProductInfo, Arrays.asList(productId), Arrays.asList(taskKeyPrefix, productKeyPrefix))
                .last()
                .map(resp -> {
                    List<Long> res = (List<Long>) resp;

                    long concurrencies = res.get(0),
                            queue_size = res.get(1),
                            priority = res.get(2),
                            wip = res.get(4),
                            inprogress = res.get(5);

                    if (inprogress >= concurrencies)
                        priority = -1;

                    task.setPriority((int) priority);

                    if (wip > queue_size)
                        throw new StaticError(Errors.ERR_TASK_OVER_QUEUE);
                    else
                        return res;
                });
    }
}
